package org.huangrui.spark.scala.streaming

import org.apache.spark.SparkConf
import org.apache.spark.streaming.dstream.{DStream, ReceiverInputDStream}
import org.apache.spark.streaming.{Seconds, StreamingContext}

/**
 * 恢复
 * @Author hr
 * @Create 2024-10-22 3:43 
 */
object SparkStreaming09_Resume {
  def main(args: Array[String]): Unit = {
    val sc: StreamingContext = StreamingContext.getActiveOrCreate("cp", () => {
      val sparkConf: SparkConf = new SparkConf().setMaster("local[*]").setAppName("SparkStreaming")
      val sc: StreamingContext = new StreamingContext(sparkConf, Seconds(3))

      val lines: ReceiverInputDStream[String] = sc.socketTextStream("localhost", 9999)
      val wordToOne: DStream[(String, Int)] = lines.flatMap(_.split(" ")).map((_, 1))

      wordToOne.print()
      sc
    })

    sc.checkpoint("cp")

    sc.start()
    sc.awaitTermination() // block 阻塞main线程
  }
}
